package com.group1.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.app.BaseAppV1;
import com.group1.realtime.bean.UserBackFlowBean;
import com.group1.realtime.util.AtguiguUtil;
import com.group1.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

import static com.group1.realtime.common.Constant.*;

public class Dws_UserBackFlow extends BaseAppV1 {

    public static void main(String[] args) {
        new Dws_UserBackFlow().init(
                3011,
                2,
                "Dws_UserBackFlow",
                TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //过滤
        SingleOutputStreamOperator<UserBackFlowBean> backFlowStream = etl(stream);
//        backFlowStream.print();
        //开窗聚合
        SingleOutputStreamOperator<UserBackFlowBean> aggregatedStream = windowAndAgg(backFlowStream);
        //aggregatedStream.print();
        // 3. 把结果写入到clickHouse中
        writeToClickHouse(aggregatedStream);

    }

    private void writeToClickHouse(SingleOutputStreamOperator<UserBackFlowBean> aggregatedStream) {
        aggregatedStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_user_backFlow_window", UserBackFlowBean.class));
    }

    private SingleOutputStreamOperator<UserBackFlowBean> windowAndAgg(SingleOutputStreamOperator<UserBackFlowBean> backFlowStream) {

        return backFlowStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<UserBackFlowBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                        .withIdleness(Duration.ofSeconds(5))
                ).windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<UserBackFlowBean>("late") {})
                .reduce(
                        new ReduceFunction<UserBackFlowBean>() {
                            @Override
                            public UserBackFlowBean reduce(UserBackFlowBean value1, UserBackFlowBean value2) throws Exception {
                                value1.setBackFlowCt(value1.getBackFlowCt() + value2.getBackFlowCt());
                                //test
                                //System.out.println(value1.toString());
                                return value1;
                            }
                        },
                        new ProcessAllWindowFunction<UserBackFlowBean, UserBackFlowBean, TimeWindow>() {
                            @Override
                            public void process(Context ctx, Iterable<UserBackFlowBean> elements, Collector<UserBackFlowBean> collector) throws Exception {

                                UserBackFlowBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                                bean.setTs(System.currentTimeMillis());
                                collector.collect(bean);
                            }
                        }
                );
    }

    private SingleOutputStreamOperator<UserBackFlowBean> etl(DataStreamSource<String> stream) {

        return stream.map(JSON::parseObject)
                .keyBy(object -> object.getJSONObject("common").getString("uid"))
                .process(new KeyedProcessFunction<String, JSONObject, UserBackFlowBean>() {

                    private ValueState<Long> loginState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        loginState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("loginState", Long.class));
                    }

                    @Override
                    public void processElement(JSONObject object, Context context, Collector<UserBackFlowBean> collector) throws Exception {
                        //test
                        //System.out.println(object.toJSONString());
                        Long ts = object.getLong("ts");
                        if (loginState.value() == null){
                            loginState.update(ts);
                        }else if (ts-loginState.value()> 3600 * 24 * 7 * 1000){
                            //test
                            //System.out.println("回流："+object.getJSONObject("common").getString("uid")+ts+" - "+loginState.value());
                            collector.collect(new UserBackFlowBean("", "", 1L, ts));
                        }else {
                            loginState.update(ts);
                        }
                    }
                });

    }
}
